package com.shujia.sql

import org.apache.spark.sql.{Column, DataFrame, SparkSession}

object Code09DSL {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("dsl")
      .config("spark.sql.shuffle.partitions", "3")
      .getOrCreate()

    val stuDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      // 1500100001,施笑槐,22,女,文科六班
      .schema("id String,name String,age int,gender String,clazz String")
      .load("spark_code/data/students_join.txt")




    // sql函数
    import org.apache.spark.sql.functions._
    // Spark隐式转换
    import spark.implicits._

    val scoreDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      // 1500100001,施笑槐,22,女,文科六班
      .schema("id String,courseID String,score int")
      .load("spark_code/data/score_join.txt")

    scoreDF
      .groupBy("id")
      .agg(sum($"score") as "total_score")
      .join(stuDF, List("id")) // 对于多个相同列进行关联时，可以添加List
      .show()



    //    println(countNum)
    //      .show()

    scoreDF
      .groupBy("id")
      .agg(sum($"score") as "total_score")
      .join(stuDF, "id") // 使用相同列ID进行关联
      .show()

    scoreDF
      .groupBy($"id" as "score_id")
      .agg(sum($"score") as "total_score")
      .join(stuDF, $"id" === $"score_id") // 使用Join表达式进行关联,但是给定列不能出现重名
      .show()


    scoreDF
      .groupBy($"id" as "score_id")
      .agg(sum($"score") as "total_score")
      .join(stuDF, $"id" === $"score_id", joinType = "left") // 左连接
      .show()

    scoreDF
      .groupBy($"id" as "score_id")
      .agg(sum($"score") as "total_score")
      .join(stuDF, $"id" === $"score_id", joinType = "right") // 右连接
      .show()

    scoreDF
      .groupBy($"id" as "score_id")
      .agg(sum($"score") as "total_score")
      .join(stuDF, $"id" === $"score_id", joinType = "full") // 全连接
      .show()

    scoreDF
      .groupBy($"id" as "score_id")
      .agg(sum($"score") as "total_score")
      .join(stuDF, expr("id = score_id"), joinType = "full") // 全连接
      .show()


    // df.join(df, df("key") === df("key"))
    scoreDF
      .groupBy($"id")
      .agg(sum($"score") as "total_score")
      .join(stuDF, scoreDF("id") === stuDF("id")  , joinType = "full") // 全连接
      .limit(1)
      .show(1)


  }
}
